package com.navinfo.opentsp.platform.computing.serializer;

import com.navinfo.opentsp.platform.computing.util.DateUtils;
import com.navinfo.opentsp.platform.location.protocol.common.LCIncidentAlarmData;
import com.navinfo.opentsp.platform.location.protocol.common.LCLocationData;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.conf.ComponentConfiguration;
import org.apache.flume.sink.hbase.HbaseEventSerializer;
import org.apache.hadoop.hbase.client.Increment;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Row;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

/**
 * 驾驶行为分析2.0 D009协议 序列化
 */
public class HbaseIncidentAlarmSerializer implements HbaseEventSerializer {
    private static final Logger log = LoggerFactory.getLogger(HbaseIncidentAlarmSerializer.class);
    private Event currentEvent;
    private byte[] colFam;

    public void initialize(Event event, byte[] colFam) {
        //byte[]字节型数组
        this.currentEvent = event;
        this.colFam = colFam;
    }
    public void close() {
        this.colFam = null;
        this.currentEvent = null;
    }

    public void configure(Context context) {   }
    public void configure(ComponentConfiguration conf) {    }

    public List<Increment> getIncrements() {
        return  new ArrayList<Increment>();
    }

    public List<Row> getActions() {
        try {
            LCIncidentAlarmData.IncidentAlarmData incidentAlarmData = LCIncidentAlarmData.IncidentAlarmData.parseFrom(currentEvent.getBody());

            //rowkey:reverse(tId)|3000000000L-gpsTime
            String rowKey = new StringBuilder(Long.valueOf(incidentAlarmData.getTerminalId()).toString()).reverse() +"\\x00"+String.valueOf(DateUtils.format(incidentAlarmData.getGpsTime(), "yyyyMMdd"))+"\\x00" +(3000000000L - incidentAlarmData.getGpsTime());
            List<Row> puts = new ArrayList<Row>();
            puts.add(preparedata(new Put(rowKey.getBytes()), incidentAlarmData));
            return puts;
        } catch (Exception e) {
            log.error("驾驶行为分析2.0 D009协议HBaseIncidentAlarm落盘序列化失败："+e.getMessage(), e);
        }

        return  Collections.emptyList();
    }

    private  Put preparedata(Put putReq, LCIncidentAlarmData.IncidentAlarmData incidentAlarmData) throws Exception {
        putReq.addColumn(colFam, Bytes.toBytes("tid"),Bytes.toBytes(String.valueOf(incidentAlarmData.getTerminalId())));
        putReq.addColumn(colFam, Bytes.toBytes("gpstime"),Bytes.toBytes(String.valueOf(incidentAlarmData.getGpsTime())));
        putReq.addColumn(colFam, Bytes.toBytes("day"), Bytes.toBytes(String.valueOf(DateUtils.format(incidentAlarmData.getGpsTime(), "yyyyMMdd"))));
        putReq.addColumn(colFam, Bytes.toBytes("serial_number"),Bytes.toBytes(String.valueOf(incidentAlarmData.getSerialNumber())));
        putReq.addColumn(colFam, Bytes.toBytes("receive_date"),Bytes.toBytes(String.valueOf(incidentAlarmData.getReceiveDate())));
        putReq.addColumn(colFam, Bytes.toBytes("latitude"),Bytes.toBytes(String.valueOf(incidentAlarmData.getLatitude())));
        putReq.addColumn(colFam, Bytes.toBytes("longitude"),Bytes.toBytes(String.valueOf(incidentAlarmData.getLongitude())));
        putReq.addColumn(colFam, Bytes.toBytes("originalLng"),Bytes.toBytes(String.valueOf(incidentAlarmData.getOriginalLat())));
        putReq.addColumn(colFam, Bytes.toBytes("originalLat"),Bytes.toBytes(String.valueOf(incidentAlarmData.getOriginalLng())));
        putReq.addColumn(colFam, Bytes.toBytes("height"),Bytes.toBytes(String.valueOf(incidentAlarmData.getHeight())));
        putReq.addColumn(colFam, Bytes.toBytes("direction"),Bytes.toBytes(String.valueOf(incidentAlarmData.getDirection())));
        putReq.addColumn(colFam, Bytes.toBytes("depotlogo"),Bytes.toBytes(String.valueOf(incidentAlarmData.getDepotLogo())));
        putReq.addColumn(colFam, Bytes.toBytes("rapid_acceleration_duration"),Bytes.toBytes(String.valueOf(incidentAlarmData.getRapidAcceleration().getDuration())));
        putReq.addColumn(colFam, Bytes.toBytes("rapid_acceleration_fuel"),Bytes.toBytes(String.valueOf(incidentAlarmData.getRapidAcceleration().getRapidAccelerationFuelConsumption())));
        putReq.addColumn(colFam, Bytes.toBytes("rapid_acceleration_status"),Bytes.toBytes(String.valueOf(incidentAlarmData.getRapidAcceleration().getRapidStatus())));
        putReq.addColumn(colFam, Bytes.toBytes("rapid_deceleration_duration"),Bytes.toBytes(String.valueOf(incidentAlarmData.getRapidDeceleration().getDuration())));
        putReq.addColumn(colFam, Bytes.toBytes("rapid_deceleration_status"),Bytes.toBytes(String.valueOf(incidentAlarmData.getRapidDeceleration().getRapidDecelerationStatus())));
        putReq.addColumn(colFam, Bytes.toBytes("over_speed_fuel_consumption"),Bytes.toBytes(String.valueOf(incidentAlarmData.getOverSpeed().getFuelConsumption())));
        putReq.addColumn(colFam, Bytes.toBytes("over_speed_status"),Bytes.toBytes(String.valueOf(incidentAlarmData.getOverSpeed().getOverSpeedStatus())));
        putReq.addColumn(colFam, Bytes.toBytes("over_speed_duration"),Bytes.toBytes(String.valueOf(incidentAlarmData.getOverSpeed().getDuration())));
        putReq.addColumn(colFam, Bytes.toBytes("super_long_idlestatus"),Bytes.toBytes(String.valueOf(incidentAlarmData.getSuperLongIdle().getSuperLongIdleStatus())));
        putReq.addColumn(colFam, Bytes.toBytes("super_long_idleduration"),Bytes.toBytes(String.valueOf(incidentAlarmData.getSuperLongIdle().getDuration())));
        putReq.addColumn(colFam, Bytes.toBytes("super_long_idlefuel_Consumption"),Bytes.toBytes(String.valueOf(incidentAlarmData.getSuperLongIdle().getCumulativeFuelConsumption())));
        putReq.addColumn(colFam, Bytes.toBytes("idle_speed_status"),Bytes.toBytes(String.valueOf(incidentAlarmData.getIdleSpeedAirConditioner().getIdleSpeedAirConditionerStatus())));
        putReq.addColumn(colFam, Bytes.toBytes("idle_speed_duration"),Bytes.toBytes(String.valueOf(incidentAlarmData.getIdleSpeedAirConditioner().getDuration())));
        putReq.addColumn(colFam, Bytes.toBytes("idle_speed_fuel"),Bytes.toBytes(String.valueOf(incidentAlarmData.getIdleSpeedAirConditioner().getCumulativeFuelConsumption())));
        putReq.addColumn(colFam, Bytes.toBytes("big_throttle_status"),Bytes.toBytes(String.valueOf(incidentAlarmData.getBigThrottle().getBigThrottleStatus())));
        putReq.addColumn(colFam, Bytes.toBytes("big_throttle_duration"),Bytes.toBytes(String.valueOf(incidentAlarmData.getBigThrottle().getDuration())));
        putReq.addColumn(colFam, Bytes.toBytes("big_throttle_fuel"),Bytes.toBytes(String.valueOf(incidentAlarmData.getBigThrottle().getCumulativeFuelConsumption())));
        putReq.addColumn(colFam, Bytes.toBytes("neutral_taxi_status"),Bytes.toBytes(String.valueOf(incidentAlarmData.getNeutralTaxi().getNeutralTaxiStatus())));
        putReq.addColumn(colFam, Bytes.toBytes("neutral_taxi_duration"),Bytes.toBytes(String.valueOf(incidentAlarmData.getNeutralTaxi().getDuration())));
        putReq.addColumn(colFam, Bytes.toBytes("fatigue_driving_time"),Bytes.toBytes(String.valueOf(incidentAlarmData.getFatigueDriving().getFatigueDrivingTime())));
        putReq.addColumn(colFam, Bytes.toBytes("fatigue_driving_status"), Bytes.toBytes(String.valueOf(incidentAlarmData.getFatigueDriving().getFatigueDrivingStatus())));

        return putReq;
    }
}
